package org.example.message_queue.mqserver.datacenter;


import org.example.message_queue.common.MqException;
import org.example.message_queue.mqserver.core.Binding;
import org.example.message_queue.mqserver.core.Exchange;
import org.example.message_queue.mqserver.core.MSGQueue;
import org.example.message_queue.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
*   同一管理内存中的所有数据
*   使用hashMap时要考虑线程安全问题
*/
public class MemoryDataCenter {
//    exchageMap：key 是 exchangeName,value 是 Exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
//    queueMap：key 是 queueName,value 是 queue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
//    bindingMap：key 是 exchangeName,value 是 queueName 对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
//    messageMap：key 是 messageName,value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
//    queueMessageMap：key是queueName，value是message的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
//    queueMessageWaitAckMap：key是queueName，key是messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter]新交换机添加成功！exchangeName="+exchange.getName());
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter]交换机删除成功！exchangeName="+exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter]新队列添加成功！queueName="+queue.getName());
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter]队列删除成功！queueName="+queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
//    先用exchangeName查看是否存在哈希表，若不存在则创建
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchageName());
//        if (bindingMap==null){
////            创建新的哈希表
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchageName(),bindingMap);
//        }
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
//      先检查一下queue是否存在,如果存在，则抛出异常，不存在则插入
        synchronized (bindingMap){
            if (bindingMap.get(binding.getQueueName())!=null){
                throw new MqException("[MemoryDataCenter]绑定已经存在！exchangeName="+binding.getExchangeName()+",queueName="+binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter]新的绑定成功！exchangeName="+binding.getExchangeName()+",queueName="+binding.getQueueName());
    }

//    public void insertbinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String,Binding> bindingmap = bindingsMap.get(binding.getExchageName());
//        if (bindingmap == null){
//            bindingmap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchageName(),bindingmap);
//        }
//        synchronized (bindingmap){
//            if (bindingmap.get(binding.getQueueName()) != null){
//                throw new MqException("[MemoryDataCenter]绑定已经存在！exchangeName="+binding.getExchageName()+",queueName="+binding.getQueueName());
//            }
//            bindingmap.put(binding.getQueueName(),binding);
//        }
//    }

//        获取绑定，两个版本
//        1.根据exchangeName和queueName确定唯一一个Binding
//        2.根据exchangeName，获取所有的Binding
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null){
//            说明该交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter]绑定不存在！exchangeName="+binding.getExchangeName()+",queueName="+binding.getQueueName());
        }
//        存在，就进行删除操作
        bindingMap.remove(binding.getQueueName());
    }

/*
* 消息的管理
*/
//    添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter]新消息添加成功！messageId="+message.getMessageId());
    }

//    根据id查询消息
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

//    根据id删除消息
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter]消息移除成功！messageId="+messageId);
    }

//    发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
//        把消息方法哦对应的数据结构中
//        1.先根据队列的名字，找到该队列对应的消息链表
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if (message == null){
//            messages = new LinkedList<>();
//            queueMessageMap.put(queue.getName(),messages);
//        }
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
//           也往消息中心插入消息
        addMessage(message);
        System.out.println("[MemoryDataCenter]消息投递到队列中成功！messageId="+message.getMessageId());
    }


//    从队列中取消息
    public Message pollMessage(String queueName){
//        根据队列名，查找一下，对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null || messages.size() == 0){
            return null;
        }
        synchronized (messages){
//        链表中有元素，则进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter]从消息队列中取出消息！messageId="+currentMessage.getMessageId());
            return currentMessage;
        }
    }

//    获取指定队列中的消息个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

//    添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter]消息进入待确认队列！messageId="+message.getMessageId());
    }

//    删除未确认的消息
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null){
            return;
        }
        messageHashMap.remove(messageId);
    }

//    获取未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

//    从硬盘上读取所有数据，把硬盘中之前持久化存储的各个维度的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
//        0.清空之前所有的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

//        List<Exchange> exchanges2 = diskDataCenter.selectAllExchanges();
//        for (Exchange exchange : exchanges2){
//            exchangeMap.put(exchange.getName(),exchange);
//        }
//        1.恢复所有的交换队列数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
//        放入到exchangeMap中
        for(Exchange exchange: exchanges ){
            exchangeMap.put(exchange.getName(),exchange);
        }

//        2.恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue: queues){
            queueMap.put(queue.getName(),queue);
        }

//        List<MSGQueue> queues2 = diskDataCenter.selectAllQueues();
//        for (MSGQueue queue : queues2){
//            queueMap.put(queue.getName(),queue);
//        }

//        3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingsMap.put(binding.getQueueName(),bindingMap);
        }

//        List<Binding> bindings2 = diskDataCenter.selectAllBindings();
//        for (Binding binding : bindings2){
////            绑定插入之前要根据exchangename来查找表换机是否存在，再根据queuename插入binding
//            ConcurrentHashMap<String,Binding> bindingMap2 = bindingsMap.computeIfAbsent(binding.getExchageName(),k-> new ConcurrentHashMap<>());
//            bindingMap2.put(binding.getQueueName(),binding);
//        }

//        4.恢复所有的消息数据
//        遍历所有队列，根据每个队列的名字，获取到所有的消息
        for (MSGQueue queue : queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages){
                messageMap.put(message.getMessageId(),message);
            }
        }


//        5.关于待确认消息的加载是不需要的，一旦在等待ack的过程中，服务器重启了，这个时候“未被确认的消息”，会恢复为“未被取走的消息”


    }


}
